package panda.query.scan;

import java.util.ArrayList;
import java.util.List;

import panda.query.struct.Attribute;
import panda.query.struct.Constant;
import panda.query.struct.NullConstant;
import panda.record.TupleToken;

public class MergeJoinScanner implements Scanner {
	UpdatableScanner l, r;
	List<Attribute> la, ra;
	List<Constant> currentVal;
	TupleToken markup = null;
	
	public MergeJoinScanner(UpdatableScanner l, UpdatableScanner r, List<Attribute> la, List<Attribute> ra) {
		this.l = l;
		this.r = r;
		this.la = la;
		this.ra = ra;
		currentVal = new ArrayList<Constant>(la.size());
		for (int i = 0; i < la.size(); i++) currentVal.add(null);
		init();
	}
	
	private int compare() {
		int res;
		
		for (int i = 0; i < la.size(); i++) {
			Constant a = l.getValue(la.get(i));
//			System.out.print(l.getValue(la.get(i)).getContentValue() + " , ");
			Constant b = r.getValue(ra.get(i));
//			System.out.println(r.getValue(ra.get(i)).getContentValue());
			
			res = a.compareTo(b);
			
			if (res != 0) return res;
		}
		return 0;
	}
	
	/**
	 * Check whether the scanner points to a tuple that equals to the current
	 * value and whether there exists <i>NULL</i> in the tuple 
	 * @param x
	 * @param attrs
	 * @return 1 if equals, 0 <i>NULL</i> found, -1 not equals
	 */
	private int isEqual(UpdatableScanner x, List<Attribute> attrs) {
		if (currentVal.get(0) == null) return -1;
		for (int i = 0; i < attrs.size(); i++) {
			Constant a = x.getValue(attrs.get(i));
			if (a instanceof NullConstant)
				return 0;
			if (a.compareTo(currentVal.get(i)) != 0)
				return -1;
		}
		return 1;
	}
	
	private void saveCurrentValue() {
		for (int i = 0; i < la.size(); i++)
			currentVal.set(i, l.getValue(la.get(i)));	
	}
	
	public void init() {
		l.init();
		r.init();
	}

	public void close() {
		l.close();
		r.close();
	}

	public boolean next() {
		boolean hasNextL, hasNextR;
		int tRes;
		
		hasNextR = r.next();
		
		if (hasNextR) {
			tRes = isEqual(r, ra);
			if (tRes == 0) hasNextR = false;
			if (hasNextR && tRes == 1)
				return true;
		}
		
		hasNextL = l.next();
		if (hasNextL) {
			tRes = isEqual(l, la);
			if (tRes == 0) hasNextL = false;
			if (hasNextL && tRes == 1) {
				r.setCursor(markup);
				return true;
			}
		}
		
		while (hasNextL && hasNextR) {
			int res = compare();
			
			if (Math.abs(res) == Integer.MAX_VALUE)
				return false;
			else if (res < 0)
				hasNextL = l.next();
			else if (res > 0)
				hasNextR = r.next();
			else {
				markup = r.getCursor();
				saveCurrentValue();
				return true;
			}
		}
		return false;
	}
	
	public Constant getValue(Attribute attr) {
		if (l.hasAttribute(attr))
			return l.getValue(attr);
		else
			return r.getValue(attr);
	}

	public boolean hasAttribute(Attribute name) {
		return l.hasAttribute(name) || r.hasAttribute(name);
	}

}
